package com.jcj.jcjmq.server;

import com.jcj.jcjmq.model.Message;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class MessageQueue {

    private String topic;
    private Message<?>[] queue = new Message[1024 * 10];
    private Integer index = 0;

    public static final Map<String, MessageQueue> queues = new HashMap<>();

    static {
        queues.put("jcj-queue", new MessageQueue("jcj-queue"));
        queues.put("a", new MessageQueue("a"));
    }

    private Map<String, MessageSubscription> subscriptions = new HashMap<>();

    public MessageQueue(String topic) {
        this.topic = topic;
    }

    public int send(Message<?> message) {
        if (index >= queue.length) {
            return -1;
        }
        message.getHeaders().put("x-offset", String.valueOf(index));
        queue[index++] = message;
        return index;
    }

    public Message<?> recv(int index) {
        if (index < this.index) {
            return queue[index];
        }
        return null;
    }

    public void subscribe(MessageSubscription subscription) {
        if (subscriptions.containsKey(subscription.getConsumerId())) {
            return;
        }
        subscriptions.put(subscription.getConsumerId(), subscription);
    }

    public void unsubscribe(MessageSubscription subscription) {
        if (subscriptions.containsKey(subscription.getConsumerId())) {
            return;
        }
        subscriptions.remove(subscription.getConsumerId());
    }

    public static void sub(MessageSubscription subscription) {
        String topic = subscription.getTopic();
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found");
        }
        messageQueue.subscribe(subscription);
    }

    public static void unsub(MessageSubscription subscription) {
        String topic = subscription.getTopic();
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            return;
        }
        messageQueue.unsubscribe(subscription);
    }

    public static int send(String topic, Message<?> message) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found");
        }
        return messageQueue.send(message);
    }

    public static Message<?> recv(String topic, String consumerId) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found");
        }
        if (!messageQueue.subscriptions.containsKey(consumerId)) {
            throw new RuntimeException("consumerId not found");
        }
        //需要更新offset
        int index = messageQueue.subscriptions.get(consumerId).getOffset();
        return messageQueue.recv(index + 1);
    }


    public static List<Message<?>> batchRecv(String topic, String consumerId, Integer size) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found");
        }
        if (!messageQueue.subscriptions.containsKey(consumerId)) {
            throw new RuntimeException("consumerId not found");
        }
        int offset = messageQueue.subscriptions.get(consumerId).getOffset();
        offset += 1;
        List<Message<?>> list = new ArrayList<>();
        Message<?> recv = messageQueue.recv(offset);
        while (recv != null) {
            list.add(recv);
            if (list.size() >= size) {
                break;
            }
            recv = messageQueue.recv(offset);
        }
        log.info("batch size:{}", list.size());
        return list;
    }

    public static int ack(String topic, String consumerId, Integer offset) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) {
            throw new RuntimeException("topic not found");
        }
        if (!messageQueue.subscriptions.containsKey(consumerId)) {
            throw new RuntimeException("consumerId not found");
        }

        MessageSubscription messageSubscription = messageQueue.subscriptions.get(consumerId);

        if (offset >= messageSubscription.getOffset() && offset <= messageQueue.index) {
            log.info("ack offset:{}", offset);
            messageSubscription.setOffset(offset);
            return offset;
        } else {
            return -1;
        }
    }
}
